package com.shu.imp;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import com.shu.es.EsQueryData;

public class MyMoveData {

	public static void main(String[] args) throws Exception {
		RestHighLevelClient client = EsQueryData.getClient("127.0.0.1",9200); 
		RestHighLevelClient txClient = EsQueryData.getClient("127.0.0.1",9222); 
		
		//MyMoveData.doData(client,txClient,"weight_user*");
		MyMoveData.doData(client,txClient,"weighthistory_day*");
		
	}
	
	public static SearchSourceBuilder AppDay(SearchSourceBuilder searchSourceBuilder) throws Exception {
		List<String> list = MyMoveData.loadParam();
		searchSourceBuilder.query(QueryBuilders.boolQuery().must(QueryBuilders.termsQuery("app_name", list)));
		searchSourceBuilder.size(10000);
		return searchSourceBuilder;
	}
	
	public static SearchSourceBuilder ScaleAddFansDay(SearchSourceBuilder searchSourceBuilder) throws Exception {
		List<String> list = MyMoveData.loadParamId();
		searchSourceBuilder.query(QueryBuilders.boolQuery().must(QueryBuilders.termsQuery("official_accounts_appid", list)));
		searchSourceBuilder.size(10000);
		return searchSourceBuilder;
	}
	
	public static SearchSourceBuilder SubscribeLog(SearchSourceBuilder searchSourceBuilder) throws Exception {
		List<String> list = MyMoveData.loadParamId();
		searchSourceBuilder.query(QueryBuilders.boolQuery().must(QueryBuilders.termsQuery("appid", list)));
		searchSourceBuilder.size(10000);
		return searchSourceBuilder;
	}
	
	public static SearchSourceBuilder WeightUser(SearchSourceBuilder searchSourceBuilder) throws Exception {
		List<String> list = MyMoveData.loadParamId();
		searchSourceBuilder.query(QueryBuilders.boolQuery().must(QueryBuilders.termsQuery("official_accounts_appid", list)));
		searchSourceBuilder.size(10000);
		return searchSourceBuilder;
	}
	
	public static SearchSourceBuilder WeightUserDay(SearchSourceBuilder searchSourceBuilder) throws Exception {
		List<String> list = MyMoveData.loadParamId();
		searchSourceBuilder.query(QueryBuilders.boolQuery().must(QueryBuilders.termsQuery("official_accounts_appid", list)));
		searchSourceBuilder.size(10000);
		return searchSourceBuilder;
	}
	
	public static void doData(RestHighLevelClient client,RestHighLevelClient txClient, String index) throws Exception {
		final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(30L));
		SearchRequest searchRequest = new SearchRequest(index);
		searchRequest.scroll(scroll);
		SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
		
		if(index.equals("scale_add_fans_day*")) {
			searchSourceBuilder = ScaleAddFansDay(searchSourceBuilder);
		}else if(index.equals("subscribe_log*")) {
			searchSourceBuilder = SubscribeLog(searchSourceBuilder);
		}else if(index.equals("add_day*")) {
			searchSourceBuilder = AppDay(searchSourceBuilder);
		}else if(index.equals("weight_user*")) {
			searchSourceBuilder = WeightUser(searchSourceBuilder);
		}else if(index.equals("weighthistory_day*")) {
			searchSourceBuilder = WeightUserDay(searchSourceBuilder);
		}else {
			System.out.println("******************************出错了***************************");
			return;
		}

		searchRequest.source(searchSourceBuilder);
		SearchResponse searchResponse = client.search(searchRequest); 
		String scrollId = searchResponse.getScrollId();
		SearchHit[] searchHits = searchResponse.getHits().getHits();
		
		
		long total = 0;
		BulkRequest request = new BulkRequest();
		for (int i = 0; i < searchHits.length; i++) {
			Map<String, Object> map = searchHits[i].getSourceAsMap();
			String tempIndex = searchHits[i].getIndex();
			request.add(new IndexRequest(tempIndex, searchHits[i].getType(), searchHits[i].getId()).source(map));
			total++;
		}
		request.timeout(TimeValue.timeValueMinutes(20));
		BulkResponse bulkResponse = txClient.bulk(request);
		
		
		
		if (bulkResponse.hasFailures()) {
			System.out.println(bulkResponse.buildFailureMessage());
		} else {
			System.out.println("========================================" + index + "==" + total + "== > success");
		}
		
		while (searchHits != null && searchHits.length > 0) {
		    SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId); 
		    scrollRequest.scroll(scroll);
		    searchResponse = client.searchScroll(scrollRequest);
		    scrollId = searchResponse.getScrollId();
		    System.out.println(scrollId);
		    searchHits = searchResponse.getHits().getHits();
		    
		    int size = searchHits.length;
		    request = new BulkRequest();
		    for(int i=0; i<size;i++) {
				Map<String, Object> map = searchHits[i].getSourceAsMap();
				String tempIndex = searchHits[i].getIndex();
				request.add(new IndexRequest(tempIndex, searchHits[i].getType(), searchHits[i].getId()).source(map));
				total ++;
		    }
		    
		    if(size>0) {
			    request.timeout(TimeValue.timeValueMinutes(20));
				bulkResponse = txClient.bulk(request);
				if(bulkResponse.hasFailures()) {
					System.out.println(bulkResponse.buildFailureMessage());
				}else {
					System.out.println("========================================"+index+"=="+total+"== > success");
				}
		    }
		}
		System.out.println("---------------"+total);
	}
	
	public static List<String> loadParamId() throws Exception {
		String pathname = "C:\\Users\\mzj\\workspace\\my.test\\src\\main\\java\\com\\shu\\imp\\cc1.txt"; // 绝对路径或相对路径都可以，这里是绝对路径，写入文件时演示相对路径  
        File filename = new File(pathname); // 要读取以上路径的input。txt文件  
        InputStreamReader reader = new InputStreamReader(  
                new FileInputStream(filename)); // 建立一个输入流对象reader  
        BufferedReader br = new BufferedReader(reader); // 建立一个对象，它把文件内容转成计算机能读懂的语言  
        String line = "";  
        line = br.readLine(); 
        Set<String> set=new HashSet<String>();
        set.add(line.replace(" ", ""));
        while (line!= null) {  
            line = br.readLine(); // 一次读入一行数据  
            System.out.println(line);
            if(line!=null && !line.equals("")) {
            	set.add(line.replace(" ", ""));
            }
        }
        return new ArrayList<String>(set);
	}
	public static List<String> loadParam() throws Exception {
		String pathname = "C:\\Users\\mzj\\workspace\\my.test\\src\\main\\java\\com\\shu\\imp\\cc.txt"; // 绝对路径或相对路径都可以，这里是绝对路径，写入文件时演示相对路径  
        File filename = new File(pathname); // 要读取以上路径的input。txt文件  
        InputStreamReader reader = new InputStreamReader(  
                new FileInputStream(filename)); // 建立一个输入流对象reader  
        BufferedReader br = new BufferedReader(reader); // 建立一个对象，它把文件内容转成计算机能读懂的语言  
        String line = "";  
        line = br.readLine(); 
        Set<String> set=new HashSet<String>();
        set.add(line.replace(" ", ""));
        while (line!= null) {  
            line = br.readLine(); // 一次读入一行数据  
            System.out.println(line);
            if(line!=null && !line.equals("")) {
            	set.add(line.replace(" ", ""));
            }
        }
        return new ArrayList<String>(set);
	}
}
